Langchain 怎么使用
LangChain 简介
LangChain 是一个强大的框架，专为构建基于大语言模型（LLM）的应用程序而设计。它提供了一套完整的工具链，帮助开发者轻松集成各种 AI 模型、数据源和外部工具，构建智能化的应用程序。
安装和基础配置
首先安装 LangChain 及其相关依赖：
# 安装核心包
pip install langchain
# 安装 OpenAI 支持
pip install langchain-openai
# 安装向量数据库支持
pip install chromadb faiss-cpu
# 安装文档加载器
pip install pypdf docx2txt
# 安装完整版（包含所有可选依赖）
pip install langchain[all]
基础配置示例：
import os
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage, SystemMessage
# 设置 API 密钥
os.environ["OPENAI_API_KEY"] = "your-api-key-here"
# 初始化语言模型
llm = ChatOpenAI(
model="gpt-3.5-turbo",
temperature=0.7,
max_tokens=1000
)
# 简单对话示例
messages = [
SystemMessage(content="你是一个有用的AI助手"),
HumanMessage(content="请介绍一下 Python 的特点")
]
response = llm(messages)
print(response.content)
提示模板（Prompts）
提示模板是 LangChain 的基础组件，用于标准化和参数化与 LLM 的交互：
基础提示模板：
from langchain.prompts import PromptTemplate, ChatPromptTemplate
from langchain.prompts.chat import SystemMessagePromptTemplate, HumanMessagePromptTemplate
# 简单提示模板
simple_template = PromptTemplate(
input_variables=["product", "language"],
template="请为 {product} 写一个 {language} 的产品描述"
)
prompt = simple_template.format(product="智能手表", language="中文")
print(prompt)
# 聊天提示模板
chat_template = ChatPromptTemplate.from_messages([
SystemMessagePromptTemplate.from_template(
"你是一个专业的 {domain} 专家"
),
HumanMessagePromptTemplate.from_template(
"请解释一下 {concept} 的核心原理"
)
])
messages = chat_template.format_messages(
domain="机器学习",
concept="神经网络"
)
条件提示模板：
from langchain.prompts import ConditionalPromptTemplate
# 根据条件选择不同的提示
def get_prompt_selector(user_type):
if user_type == "beginner":
return PromptTemplate(
input_variables=["topic"],
template="请用简单的语言解释 {topic}，适合初学者理解"
)
else:
return PromptTemplate(
input_variables=["topic"],
template="请深入分析 {topic} 的技术细节和实现原理"
)
# 使用示例
beginner_prompt = get_prompt_selector("beginner")
expert_prompt = get_prompt_selector("expert")
链式操作（Chains）
Chains 是 LangChain 的核心概念，用于将多个组件连接成复杂的处理流程：
LLM Chain - 最基础的链：
from langchain.chains import LLMChain
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
# 创建 LLM Chain
llm = ChatOpenAI(temperature=0.7)
prompt = PromptTemplate(
input_variables=["company", "product"],
template="为 {company} 公司的 {product} 产品写一个营销口号"
)
chain = LLMChain(llm=llm, prompt=prompt)
# 运行链
result = chain.run(company="科技创新", product="智能机器人")
print(result)
Sequential Chain - 顺序链：
from langchain.chains import SequentialChain
# 第一个链：生成故事大纲
outline_prompt = PromptTemplate(
input_variables=["theme"],
template="为主题 '{theme}' 创建一个故事大纲"
)
outline_chain = LLMChain(llm=llm, prompt=outline_prompt, output_key="outline")
# 第二个链：根据大纲写故事
story_prompt = PromptTemplate(
input_variables=["outline"],
template="根据以下大纲写一个完整的故事：\n{outline}"
)
story_chain = LLMChain(llm=llm, prompt=story_prompt, output_key="story")
# 第三个链：为故事写摘要
summary_prompt = PromptTemplate(
input_variables=["story"],
template="为以下故事写一个简短摘要：\n{story}"
)
summary_chain = LLMChain(llm=llm, prompt=summary_prompt, output_key="summary")
# 组合成顺序链
overall_chain = SequentialChain(
chains=[outline_chain, story_chain, summary_chain],
input_variables=["theme"],
output_variables=["outline", "story", "summary"],
verbose=True
)
# 运行完整流程
result = overall_chain({"theme": "人工智能与人类友谊"})
记忆机制（Memory）
Memory 组件允许链在多轮对话中保持上下文：
对话缓冲记忆：
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
# 创建记忆组件
memory = ConversationBufferMemory()
# 创建对话链
conversation = ConversationChain(
llm=llm,
memory=memory,
verbose=True
)
# 多轮对话
print(conversation.predict(input="我叫张三，是一名软件工程师"))
print(conversation.predict(input="我的职业是什么？"))
print(conversation.predict(input="我的名字是什么？"))
# 查看记忆内容
print(memory.buffer)
对话摘要记忆：
from langchain.memory import ConversationSummaryBufferMemory
# 使用摘要记忆（节省 token）
summary_memory = ConversationSummaryBufferMemory(
llm=llm,
max_token_limit=100, # 超过限制时自动摘要
return_messages=True
)
conversation_with_summary = ConversationChain(
llm=llm,
memory=summary_memory,
verbose=True
)
# 长对话会自动摘要早期内容
for i in range(5):
response = conversation_with_summary.predict(
input=f"这是第 {i+1} 轮对话，请记住这个数字"
)
print(f"轮次 {i+1}: {response}")
文档加载和向量存储
LangChain 提供了丰富的文档加载器和向量存储支持，用于构建知识问答系统：
文档加载和处理：
from langchain.document_loaders import TextLoader, PyPDFLoader, DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
# 加载文档
loader = TextLoader("data/document.txt", encoding="utf-8")
documents = loader.load()
# 文本分割
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
length_function=len,
)
texts = text_splitter.split_documents(documents)
# 创建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(
documents=texts,
embedding=embeddings,
persist_directory="./chroma_db"
)
# 相似度搜索
query = "什么是机器学习？"
docs = vectorstore.similarity_search(query, k=3)
for doc in docs:
print(doc.page_content[:200])
构建问答系统：
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
# 自定义问答提示模板
qa_template = """使用以下上下文来回答问题。如果你不知道答案，就说你不知道，不要编造答案。
上下文: {context}
问题: {question}
回答:"""
QA_PROMPT = PromptTemplate(
template=qa_template,
input_variables=["context", "question"]
)
# 创建问答链
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
chain_type_kwargs={"prompt": QA_PROMPT},
return_source_documents=True
)
# 询问问题
question = "文档中提到了哪些机器学习算法？"
result = qa_chain({"query": question})
print("答案:", result["result"])
print("来源文档数量:", len(result["source_documents"]))
智能代理（Agents）
Agents 是 LangChain 的高级功能，能够根据用户输入决定使用哪些工具来完成任务：
创建带工具的代理：
from langchain.agents import initialize_agent, Tool, AgentType
from langchain.tools import DuckDuckGoSearchRun
from langchain.utilities import PythonREPL
# 定义工具
search = DuckDuckGoSearchRun()
python_repl = PythonREPL()
tools = [
Tool(
name="搜索",
func=search.run,
description="当需要搜索实时信息时使用"
),
Tool(
name="Python代码执行",
func=python_repl.run,
description="当需要执行 Python 代码进行计算时使用"
)
]
# 初始化代理
agent = initialize_agent(
tools=tools,
llm=llm,
agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
verbose=True,
handle_parsing_errors=True
)
# 使用代理
response = agent.run("请搜索2024年最新的AI技术趋势，然后计算一下如果投资10万元，年收益率15%，5年后能得到多少钱？")
print(response)
自定义工具：
from langchain.tools import BaseTool
from typing import Optional, Type
from pydantic import BaseModel, Field
class WeatherInput(BaseModel):
city: str = Field(description="要查询天气的城市名称")
class WeatherTool(BaseTool):
name = "天气查询"
description = "用于查询指定城市的天气信息"
args_schema: Type[BaseModel] = WeatherInput
def _run(self, city: str) -> str:
# 这里应该调用实际的天气API
return f"{city}的天气：晴天，温度25°C，湿度60%"
async def _arun(self, city: str) -> str:
"""异步版本"""
return self._run(city)
# 使用自定义工具
weather_tool = WeatherTool()
tools_with_weather = tools + [weather_tool]
weather_agent = initialize_agent(
tools=tools_with_weather,
llm=llm,
agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
verbose=True
)
result = weather_agent.run("北京的天气怎么样？如果温度是25度，请计算对应的华氏温度。")
实际应用场景
智能客服系统
from langchain.memory import ConversationSummaryBufferMemory
from langchain.chains import ConversationChain
class CustomerServiceBot:
def __init__(self):
self.llm = ChatOpenAI(temperature=0.3)
self.memory = ConversationSummaryBufferMemory(
llm=self.llm,
max_token_limit=500
)
# 客服专用提示模板
self.prompt = PromptTemplate(
input_variables=["history", "input"],
template="""你是一个专业的客服代表，请遵循以下原则：
1. 友好、耐心、专业
2. 如果不确定答案，请说"我需要为您转接专业人员"
3. 记住之前的对话内容
4. 提供具体、有用的建议
对话历史：{history}
客户：{input}
客服："""
)
self.chain = ConversationChain(
llm=self.llm,
memory=self.memory,
prompt=self.prompt,
verbose=False
)
def chat(self, user_input: str) -> str:
return self.chain.predict(input=user_input)
# 使用示例
bot = CustomerServiceBot()
print(bot.chat("我的订单什么时候能到？"))
print(bot.chat("订单号是12345"))
print(bot.chat("可以退货吗？"))
代码生成助手
class CodeAssistant:
def __init__(self):
self.llm = ChatOpenAI(model="gpt-4", temperature=0.1)
self.code_prompt = PromptTemplate(
input_variables=["language", "description", "requirements"],
template="""作为一个专业的程序员，请根据以下要求生成代码：
编程语言：{language}
功能描述：{description}
具体要求：{requirements}
请提供：
1. 完整的代码实现
2. 代码注释
3. 使用示例
4. 注意事项
代码："""
)
self.chain = LLMChain(llm=self.llm, prompt=self.code_prompt)
def generate_code(self, language: str, description: str, requirements: str = "") -> str:
return self.chain.run(
language=language,
description=description,
requirements=requirements
)
# 使用示例
assistant = CodeAssistant()
code = assistant.generate_code(
language="Python",
description="实现一个简单的Web爬虫",
requirements="使用requests库，支持错误处理，能够解析HTML"
)
print(code)
性能优化和最佳实践
成本和性能优化：
from langchain.cache import InMemoryCache
from langchain.globals import set_llm_cache
import asyncio
# 启用缓存
set_llm_cache(InMemoryCache())
# 批量处理示例
async def batch_process_queries(queries: list) -> list:
tasks = []
for query in queries:
task = asyncio.create_task(process_single_query(query))
tasks.append(task)
results = await asyncio.gather(*tasks)
return results
async def process_single_query(query: str) -> str:
# 异步处理单个查询
chain = LLMChain(llm=llm, prompt=prompt)
return await chain.arun(input=query)
# 使用示例
queries = ["什么是AI?", "Python的优势", "机器学习应用"]
results = asyncio.run(batch_process_queries(queries))
错误处理和监控：
import logging
from langchain.callbacks import get_openai_callback
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def robust_chain_execution(chain, input_data, max_retries=3):
for attempt in range(max_retries):
try:
with get_openai_callback() as cb:
result = chain.run(input_data)
# 记录使用情况
logger.info(f"Token使用: {cb.total_tokens}, 成本: ${cb.total_cost:.4f}")
return result
except Exception as e:
logger.error(f"尝试 {attempt + 1} 失败: {str(e)}")
if attempt == max_retries - 1:
raise e
time.sleep(2 ** attempt) # 指数退避
return None
LangChain 为构建智能应用提供了强大而灵活的框架。通过合理使用其各种组件，我们可以快速构建出功能丰富的 AI 应用程序。关键是要根据具体需求选择合适的组件组合，并注意性能优化和成本控制。